package homework.week5.consistencyhash;

import com.alibaba.fastjson.JSONObject;
import homework.util.Print;

import java.util.*;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 一致性hash
 * 参考来源：https://www.cnblogs.com/williamjie/p/9477852.html
 * <p>
 * 0~2^32-1的虚拟圆:一致性Hash环
 * 1、平衡性(Balance)：平衡性是指哈希的结果能够尽可能分布到所有的缓冲中去，这样可以使得所有的缓冲空间都得到利用
 * 2、单调性(Monotonicity)：单调性是指如果已经有一些内容通过哈希分派到了相应的缓冲中，又有新的缓冲区加入到系统中，那么哈希的结果应能够保证原有已分配的内容可以被映射到新的缓冲区中去，而不会被映射到旧的缓冲集合中的其他缓冲区
 * 3、分散性(Spread)：在分布式环境中，终端有可能看不到所有的缓冲，而是只能看到其中的一部分。当终端希望通过哈希过程将内容映射到缓冲上时，由于不同终端所见的缓冲范围有可能不同，从而导致哈希的结果不一致，最终的结果是相同的内容被不同的终端映射到不同的缓冲区中。这种情况显然是应该避免的，因为它导致相同内容被存储到不同缓冲中去，降低了系统存储的效率。分散性的定义就是上述情况发生的严重程度。好的哈希算法应能够尽量避免不一致的情况发生，也就是尽量降低分散性。
 * 4、负载(Load)：负载问题实际上是从另一个角度看待分散性问题。既然不同的终端可能将相同的内容映射到不同的缓冲区中，那么对于一个特定的缓冲区而言，也可能被不同的用户映射为不同的内容。
 * 5、平滑性(Smoothness)：平滑性是指缓存服务器的数目平滑改变和缓存对象的平滑改变是一致的。
 */
public class ConsistencyHash {

    // 物理节点列表
    public final List<String> physicalIpList;

    // 虚拟节点个数
    public final AtomicInteger virtualNodeSize;

    // 虚拟节点连接符
    private final static String CONNECTOR_NAME = "-Node";

    // 虚拟节点缓存
    private final ConcurrentNavigableMap<Integer, String> nodeCache = new ConcurrentSkipListMap<>();

    public ConsistencyHash(final List<String> physicalIpList, int virtualNodeSize) {
        this.physicalIpList = physicalIpList;
        this.virtualNodeSize = new AtomicInteger(virtualNodeSize);
        // 初始化
        loadVirtualNode(physicalIpList);
    }

    /**
     * 获取对于的物理节点Ip
     *
     * @param key 缓存内容的key
     * @return 物理节点ip
     */
    public String getPhysicalIp(String key) {
        // 缓存内容key的hashCode
        int keyHashCode = calculateNode(key);
        // 从hash环中获取虚拟节点
        ConcurrentNavigableMap<Integer, String> tailMap = nodeCache.tailMap(keyHashCode);
        // 判断是否有虚拟节点存在，不存在则从第一个物理节点取值
        String virtualNode = !tailMap.isEmpty() ? tailMap.firstEntry().getValue() : nodeCache.firstEntry().getValue();
        // 虚拟节点映射到物理节点
        return virtualNode.split(CONNECTOR_NAME)[0];
    }

    /**
     * 新增物理节点
     *
     * @param ip 物理节点ip
     */
    public void addPhysicalIp(String... ip) {
        this.loadVirtualNode(Arrays.asList(ip));
    }

    /**
     * 移除物理节点
     *
     * @param ips 物理节点ip
     */
    public void removePhysicalIp(String... ips) {
        Arrays.stream(ips)
                .parallel()
                .forEach(ip -> {
                    for (int i = 0; i < virtualNodeSize.intValue(); ++i) {
                        String virtualName = getVirtualNameByIp(ip, i);
                        int removeKey = calculateNode(virtualName);
                        nodeCache.remove(removeKey);
                    }
                });
    }


    /**
     * 节点计算
     * 使用FNV1_32_HASH算法计算服务器的Hash,使用默认的hashCode()（NATIVE_HASH）计算时，发现都被分发到了同一个物理节点上
     * 使用FNV1_32_HASH算法时，根据虚拟节点数量的不同，请求可以分配到近乎均匀的每个物理节点上
     * 参考了博客：https://blog.csdn.net/whut_gyx/article/details/39002191
     * @param key 缓存内容或者Ip
     * @return hashCode-Node
     */
    public int calculateNode(String key) {
        final int p = 16777619;
        int hash = (int) 2166136261L;
        for (int i = 0; i < key.length(); i++)
            hash = (hash ^ key.charAt(i)) * p;
        hash += hash << 13;
        hash ^= hash >> 7;
        hash += hash << 3;
        hash ^= hash >> 17;
        hash += hash << 5;

        // 如果算出来的值为负数则取其绝对值
        if (hash < 0) {
            hash = Math.abs(hash);
        }
        return hash;
    }

    /**
     * 构建虚拟IP
     *
     * @param ip
     * @param i
     * @return
     */
    private String getVirtualNameByIp(String ip, int i) {
        return ip + CONNECTOR_NAME + (++i);
    }

    /**
     * 加载节点的hashCode到缓存中
     *
     * @param physicalIp 物理节点
     */
    private void loadVirtualNode(final List<String> physicalIp) {
        // 先计算物理节点的hash
        if (null == physicalIp || physicalIp.isEmpty()) {
            throw new RuntimeException("物理节点不能为空");
        }
        // 计算物理节点的hashCode并缓存虚拟节点
        for (String ip : physicalIp) {
            // 获取物理节点
            addVirtualNode(ip, this.virtualNodeSize.intValue());
        }
    }

    /**
     * 为物理节点添加虚拟节点
     *
     * @param virtualNodeSize 待分配的虚拟节点
     * @param physicalIp      物理节点ip
     */
    private void addVirtualNode(String physicalIp, int virtualNodeSize) {
        // 储存每个物理节点映射的虚拟节点
        for (int j = 0; j < virtualNodeSize; ++j) {
            String virtualName = getVirtualNameByIp(physicalIp, j);
            nodeCache.put(calculateNode(virtualName), virtualName);
        }
    }

    /**
     * 打印缓存中的数据
     */
    public void printCache() {
        ConcurrentNavigableMap<Integer, String> nodeCache = this.nodeCache;
        Print.out("%s", JSONObject.toJSONString(nodeCache));
    }
}
